package com.sharp;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.Map;

/**
 * Hello world!
 *
 */
public class RemoteMyTopology
{

    public static class MySpout extends BaseRichSpout{

        private Map map;
        private TopologyContext topologyContext;
        private SpoutOutputCollector spoutOutputCollector;
        @Override
        public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.map = map;
            this.topologyContext = topologyContext;
            this.spoutOutputCollector = spoutOutputCollector;
        }

        int num = 0;
        @Override
        public void nextTuple() {
            this.spoutOutputCollector.emit(new Values(num++, "sharp"+num));
            Utils.sleep(2000);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("num","name"));
        }
    }

    public static class MyBlot extends BaseRichBolt{
        private Map stormConf;
        private TopologyContext context;
        private OutputCollector collector;
        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.stormConf = stormConf;
            this.context = context;
            this.collector = collector;
        }
        static Integer count = 0;
        @Override
        public void execute(Tuple input) {
            Integer num = input.getIntegerByField("num");
            String name = input.getStringByField("name");
            count += num;
            System.out.println(name+"----->"+count);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {

        }
    }
    public static void main( String[] args ) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
        TopologyBuilder topology = new TopologyBuilder();
        topology.setSpout("spout_id", new MySpout());
        topology.setBolt("bolt_id", new MyBlot()).shuffleGrouping("spout_id");
        topology.setBolt("bolt_id2", new MyBlot()).shuffleGrouping("spout_id");
        StormTopology topo = topology.createTopology();
        StormSubmitter.submitTopology(RemoteMyTopology.class.getSimpleName(), new Config(), topo);
    }
}
